package com.etlmaster.executor.steps.rowprocessor;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.DpExecutorUtils;
import com.etlmaster.executor.utils.StepState;
import com.etlmaster.jdbc.JdbcTemplate;
import com.etlmaster.rowprocessor.core.JobManager;

import freemarker.core.FreeMarkerTemplate;

public class Bolt {

	public static Map<String,List<String>> getIndex(String jobName){
		Map<String,String> data=new HashMap<String,String>();
		data.put("jobName", jobName);
		String fetchRowProcessRelation=DpExecutorUtils.variableSubstitution(DataCollectionUtils.getProperties("fetchRowProcessRelation"),data);
		JdbcTemplate jdbc=new JdbcTemplate();
		List<Map<String, Object>> list=jdbc.queryForList(fetchRowProcessRelation);
		Map<String,List<String>> index=new HashMap<String,List<String>>();
		for(Map<String,Object> map:list){
			String key=( (Integer) map.get("step_seq"))+"";
			String[] values=((String) map.get("aftid")).split(",");
			List<String> aftids=new ArrayList<String>();
			for(String value:values){
				aftids.add(value);
			}
			index.put(key, aftids);
		}
		return index;
	}
	public static int runRowProcessor(StepMeta meta,Map<String,String> map){
		String rowstep=FreeMarkerTemplate.getFpl("plugin", map);
		List<String> rowSteps=(List<String>) meta.getVariable("jobJson");
		rowSteps.add(rowstep);
		//Map<String,List<String>> indexs=JobPool.getJob((String) meta.getVariable("f")).getNextStepIndex();
		Map<String,List<String>> indexs=Bolt.getIndex((String) meta.getVariable("f"));
		Map<String,Object> data=new HashMap<String,Object>();
		data.put("indexs", indexs);
		data.put("rowsteps", rowSteps);
		data.put("jobName", (String) meta.getVariable("f"));
		String jobJson=FreeMarkerTemplate.getFpls("job", data);
		meta.log("INFO","生成{}配置成功",meta.getName());
		meta.log("INFO",jobJson);
		try {
			JobManager.startJob(jobJson);
		} catch (InterruptedException e) {
			meta.log("ERROR",e.getMessage());
			e.printStackTrace();
			return StepState.RUNNING_FAILED;
		}
		return StepState.RUNNING_SUCCESSED;
	}
}
